import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Set;

public class RelayImpl implements IRelay {
	private static final int BUFFER_SIZE = 1024;
	private static final String REQUEST_PATTERN = "\\$";

	private final String hostName;
	private final int portNumber;
	private ServerSocketChannel serverSocketChannel;
	private Selector selector;

	private Map<String, LinkedList<SocketChannel>> registration;
	private Map<String, SocketChannel> clients;

	public RelayImpl(String hostName, int portNumber) {
		this.hostName = hostName;
		this.portNumber = portNumber;
		registration = new HashMap<String, LinkedList<SocketChannel>>();
		clients = new HashMap<String, SocketChannel>();
		startRelay();
	}

	/**
	 * starting the relay by opening a socket accepting connections.
	 */
	private void startRelay() {
		try {
			selector = Selector.open();
			serverSocketChannel = ServerSocketChannel.open();
			serverSocketChannel.configureBlocking(false);
			InetSocketAddress addr = new InetSocketAddress(
					InetAddress.getByName(hostName), portNumber);
			serverSocketChannel.socket().bind(addr);
			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);				
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public void connect(String hostName, int port) {
		// TODO: implement
	}

	@Override
	public void startListening() {
		System.out.println("Start listening ...");
		Set<SelectionKey> selectedKeys;
		Iterator<SelectionKey> iterator;

		while (true) {
			try {
				selector.select();
				selectedKeys = selector.selectedKeys();
				iterator = selectedKeys.iterator();

				while (iterator.hasNext()) {
					acceptRequest(iterator);
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

	}

	private void acceptRequest(Iterator<SelectionKey> iterator) {
		SelectionKey key = iterator.next();
		iterator.remove();

		if (key.isValid() && key.isAcceptable()) {
			accept();
		}
		if (key.isValid() && key.isReadable()) {
			SocketChannel client = (SocketChannel) key.channel();
			String request = readMessage(key);
			System.out.println(request);
			String response = handleRequest(request, key);

			ByteBuffer buffer = writeMessage(client, response);
			buffer.clear();
		}
	}

	private String handleRequest(String request, SelectionKey clientKey) {
		String response = "";
		String[] requestParameters = getRequestParameters(request);
		String userName = requestParameters[0];
		int requestId = Integer.parseInt(requestParameters[1]);
		Request requestType = Request.getRequestById(requestId);

		switch (requestType) {
		case CONNECT:
			addNewClient(userName, clientKey);
			response = "You are sucessfully connected !";
			break;
		case DISCONNECT:
			removeClient(clientKey);
			response = "You are sucessfully disconnected !";
			break;
		case REGISTER:
			String topic = requestParameters[2];
			LinkedList<SocketChannel> topicRegistrations = registration
					.get(topic);
			if (topicRegistrations == null)
				topicRegistrations = new LinkedList<SocketChannel>();

			topicRegistrations.add((SocketChannel) clientKey.channel());
			registration.put(topic, topicRegistrations);
			response = "You are registered for " + topic;
			break;
		case UNREGISTER:
			topic = requestParameters[2];
			topicRegistrations = registration.get(topic);
			if (topicRegistrations == null)
				topicRegistrations = new LinkedList<SocketChannel>();

			topicRegistrations.remove((SocketChannel) clientKey.channel());
			registration.put(topic, topicRegistrations);
			response = "You are unregister from " + topic;
			break;
		case SEND:
			sendBroadcastMessage(requestParameters[2], requestParameters[3],
					(SocketChannel)clientKey.channel());
			response = "Your Message was sucessfully Forwarded !";
			break;
		default:
			response = "wrong";
		}

		return response;
	}

	private void sendBroadcastMessage(String topic, String message,
			SocketChannel from) {
		for (SocketChannel client : registration.get(topic))
			writeMessage(client, message);
	}

	private void addNewClient(String userName, SelectionKey clientKey) {
		SocketChannel clientChannel = (SocketChannel) clientKey.channel();
		clients.put(userName, clientChannel);
	}

	private void removeClient(SelectionKey clientKey) {
		SocketChannel clientChannel = (SocketChannel) clientKey.channel();
		clients.remove(clientChannel);
		clientKey.cancel();
	}

	private String[] getRequestParameters(String request) {
		return request.split(REQUEST_PATTERN);
	}

	private String readMessage(SelectionKey key) {
		String request = "";
		try {
			SocketChannel client = (SocketChannel) key.channel();
			ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
			int bytesRead = client.read(buffer);
			if (bytesRead == -1) {
				close(key);
				return "";
			}
			buffer.flip();
			byte[] input = new byte[buffer.limit()];
			buffer.get(input);
			request = new String(input);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return request;
	}

	private ByteBuffer writeMessage(SocketChannel client, String response) {
		ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
		try {
			client.write(buffer);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return buffer;
	}

	/**
	 * close client connection
	 * 
	 * @param key
	 * @throws IOException
	 */
	private void close(SelectionKey key) throws IOException {
		System.out.printf("Closing connection from client %s", key.channel());
		key.cancel();
		key.channel().close();
	}

	@Override
	public void passMessage(String message) {
		// TODO Auto-generated method stub

	}

	@Override
	public void accept() {
		SocketChannel connection;
		try {
			connection = serverSocketChannel.accept();
			connection.configureBlocking(false);
			System.out.printf("Connection from: %s has been established.\n",
					connection.getRemoteAddress().toString());
			connection.register(selector, SelectionKey.OP_READ
					| SelectionKey.OP_WRITE);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		if (args.length < 1) {
			System.out.printf("usage: java RelayImpl portNumber\n");
		} else {
			RelayImpl firstRelay = new RelayImpl(args[0],
					Integer.parseInt(args[1]));
			firstRelay.startListening();
		}
	}

}
